#ifndef __M_SERVER_H__
#define __M_SERVER_H__
#include <vector>
#include <iostream>
#include <cassert>
#include <cstring>
#include <iostream>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include "Log.hpp"
#include <arpa/inet.h>
#include <unistd.h>
#include <fcntl.h>
#include <functional>
#include <sys/epoll.h>
#include <memory>
#include <thread>
#include <mutex>
#include <sys/eventfd.h>

#include <iostream>
#include <typeinfo>
#include <cassert>
#include<any>
#include <condition_variable>
#include <signal.h>
#include <sys/timerfd.h>
#define BUFFER_DEFAULT_SIZE 1024
class Buffer
{

private:
    std::vector<char> _buffer;
    uint64_t _read_idx;
    uint64_t _write_idx;

public:
    Buffer() : _read_idx(0), _write_idx(0), _buffer(BUFFER_DEFAULT_SIZE) {}
    // 获取缓冲区首地址
    char *Begin() { return &*_buffer.begin(); }
    // 获取当前写位置地址
    char *current_write_pos()
    {
        return Begin() + _write_idx;
    }
    // 获取当前读位置地址
    char *current_read_pos() { return Begin() + _read_idx; }
    // 获取缓冲区末尾空闲空间的大小--写偏移之后的空闲空间
    uint64_t free_space()
    {
        return _buffer.size() - _write_idx;
    }
    // 获取缓冲区起始空闲空间大小--读偏移之前的空闲空间
    uint64_t start_free_space() { return _read_idx; }
    // 获取可读数据大小
    uint64_t readable_size() { return _write_idx - _read_idx; }
    // 获取可写数据大小
    uint64_t writable_size() { return _buffer.size() - _write_idx; }
    // 将读偏移向后移动
    void move_read_pos(uint64_t size)
    {
        if(size==0) return; 
        // 向后移动的大小不能超过可读数据大小
        assert(size <= readable_size());
        _read_idx += size;
    }
    // 将写偏移向后移动
    void move_write_pos(uint64_t size)
    {
        // 向后移动的大小,不能超过缓冲区剩余空间
        assert(size <= free_space());
        _write_idx += size;
    }
    // 确保可写空间足够（整体空闲空间够了就移动数据，否则就扩容）
    void ensure_writable_space(uint64_t size)
    {
        // 如果末尾空闲空间够，直接返回
        if (free_space() >= size)
        {
            return;
        }
        // 如果末尾剩余空间不够，看起始空闲空间和末尾空闲空间的大小加起是否足够
        // 够了就移动数据
        if (size <= start_free_space() + free_space())
        {
            uint64_t rsz = readable_size(); // 把当前数据的大小保存下来
            std::copy(current_read_pos(), current_read_pos() + rsz, Begin());
            _read_idx = 0;
            _write_idx = rsz;
            return;
        }
        // 扩容:总体都不够了，就扩容，不移动数据
        _buffer.resize(_buffer.size() + size);
    }
    // 写入数据
    void write(const void *data, uint64_t size)
    {
        // 确保可写空间足够、写入数据
        if(size == 0) return;
        ensure_writable_space(size);
        const char *p = (const char *)data;
        std::copy(p, p + size, current_write_pos());
        move_write_pos(size);
    }
    // 写入字符串
    void writeAsString(const std::string &str)
    {
        write(str.c_str(), str.size());
    }
    // 写入buffer
    void writeAsBuffer(Buffer &buf)
    {
        return write(buf.current_read_pos(), buf.readable_size());
    }
    // 读取数据
    void read(void *buf, uint64_t size)
    {
        // 确保我们读的大小不能超过可读数据大小
        assert(size <= readable_size());
        std::copy(current_read_pos(), current_read_pos() + size, (char *)buf);
        // move_read_pos(size);
    }
    // 读取字符串
    std::string readAsString(uint64_t size)
    {
        std::string str(size, '\0');
        read(&str[0], size);
        return str;
    }
    // 找到\n位置
    char *FindCRLF()
    {
        void *p = memchr(current_read_pos(), '\n', readable_size());
        return (char *)p;
    }
    // 取一行
    std::string readLine()
    {
        char *p = FindCRLF();
        if (p == NULL)
        {
            std::cout<<"readLine:no CRLF"<<std::endl;
            return "";
        }
        //+1是因为要把\n也读进来
        return readAsString(p - current_read_pos() + 1);
    }
    std::string GetLinePop(){
        std::string str = readLine();
        move_read_pos(str.size());
        return str;
    }
    // 清空缓冲区
    void clear()
    {
        // 直接采用覆盖的方式清空缓冲区
        _read_idx = 0;
        _write_idx = 0;
    }
};

#define MAX_LISTEN 1024
class Socket
{
private:
    int _sockfd;

public:
    Socket() : _sockfd(-1) {}
    Socket(int sockfd) : _sockfd(sockfd) {}
    ~Socket()
    {
        Close();
    }
    int GetFd() const
    {
        return _sockfd;
    }
    // 创建套接字
    bool Create()
    {
        // socket(int domain,int type,int protocol)
        _sockfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
        // AF_INET:IPv4协议，SOCK_STREAM:TCP流式套接字  IPPROTO_TCP:TCP协议
        if (_sockfd < 0)
        {
            LOG_ERROR("create socket failed");
            return false;
        }
        return true;
    }

    // 绑定地址
    bool Bind(const std::string &ip, uint16_t port)
    {
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(addr);
        // int bind(int sockfd,const struct sockaddr *addr,socklen_t addrlen)
        int ret = bind(_sockfd, (struct sockaddr *)&addr, len);
        if (ret < 0)
        {
            LOG_ERROR("bind socket failed");
            return false;
        }
        return true;
    }
    // 监听
    bool Listen(int backlog = MAX_LISTEN)
    {
        // int listen(int sockfd,int backlog)  backlog:最大连接数
        int ret = listen(_sockfd, backlog);
        if (ret < 0)
        {
            LOG_ERROR("listen socket failed");
            return false;
        }
        return true;
    }
    // 向服务器发起连接请求
    bool Connect(const std::string &ip, uint16_t port)
    {
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(addr);
        // int connect(int sockfd,const struct sockaddr *addr,socklen_t addrlen)
        int ret = connect(_sockfd, (struct sockaddr *)&addr, len);
        if (ret < 0)
        {
            LOG_ERROR("connect socket failed");
            return false;
        }
        return true;
    }
    // 获取新连接
    int Accept()
    {
        // int accept(int sockfd,struct sockaddr *addr,socklen_t *addrlen)
        // addr:客户端地址，addrlen:客户端地址长度
        int new_sockfd = accept(_sockfd, NULL, NULL);
        if (new_sockfd < 0)
        {
            LOG_ERROR("accept socket failed");
            return -1;
        }
        return new_sockfd;
    }
    // 接收数据
    ssize_t Recv(void *buf, size_t len, int flags = 0)
    {
        // ssize_t recv(int sockfd,void *buf,size_t len,int flags)
        // buf:接收缓冲区，len:接收长度，flags:接收标志
        ssize_t n = recv(_sockfd, buf, len, flags);
        if (n <= 0)
        {
            if (errno == EAGAIN || errno == EINTR)
            {
                return 0;
            }
            LOG_ERROR("recv socket failed");
            //直接退出程序
            return -1;
        }
        return n;
    }
    // 非阻塞接收数据
    ssize_t RecvNonBlock(void *buf, size_t len, int flags = 0)
    {
        // buf:接收缓冲区，len:接收长度，flags:接收标志 MSG_DONTWAIT:非阻塞
        return Recv(buf, len, MSG_DONTWAIT);
    }
    // 发送数据
    ssize_t Send(const void *buf, size_t len, int flags = 0)
    {
        // ssize_t send(int sockfd,const void *buf,size_t len,int flags)
        // buf:发送缓冲区，len:发送长度，flags:发送标志
        ssize_t n = send(_sockfd, buf, len, flags);
        if (n <= 0)
        {
            if (errno == EAGAIN || errno == EINTR)
            {
                return 0;
            }
            LOG_ERROR("send socket failed");
            return -1;
        }
        return n;
    }
    // 非阻塞发送数据
    ssize_t SendNonBlock(const void *buf, size_t len)
    {
        if(len == 0)return 0;
        // buf:发送缓冲区，len:发送长度，flags:发送标志 MSG_DONTWAIT:非阻塞
        return Send(buf, len, MSG_DONTWAIT);
    }
    // 关闭套接字
    void Close()
    {
        if (_sockfd != -1)
        {
            close(_sockfd);
            _sockfd = -1;
        }
    }
    // 创建一个服务端连接
    bool CreateServer(uint16_t port, const std::string &ip = "0.0.0.0", bool flags = false)
    {
        // 创建套接字
        if (Create() == false)
            return false;
        // 设置地址端口复用
        ReuseAddress();
        // 绑定地址
        if (Bind(ip, port) == false)
            return false;
        // 监听
        if (Listen() == false)
            return false;
        // 设置非阻塞
        if (flags)
            NonBlock();
        
        return true;
    }
    // 创建一个客户端连接
    bool CreateClient(uint16_t port, const std::string &ip)
    {
        // 创建套接字
        if (Create() == false)
            return false;
        // 连接服务器
        if (Connect(ip, port) == false)
            return false;

        return true;
    }
    // 设置套接字选项--开启地址端口复用
    void ReuseAddress()
    {
        // int setsockopt(int sockfd,int level,int optname,const void *optval,socklen_t optlen)
        int val = 1;
        // SOL_SOCKET:套接字选项，SO_REUSEADDR:地址复用选项
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, (void *)&val, sizeof(val));
        // SO_REUSEPORT:端口复用选项
        int val2 = 1;
        setsockopt(_sockfd, IPPROTO_TCP, SO_REUSEPORT, (void *)&val2, sizeof(val2));
    }
    // 设置套接字阻塞属性--设置为非阻塞
    void NonBlock()
    {
        // int fcntl(int fd,int cmd,...)
        int flags = fcntl(_sockfd, F_GETFL, 0); // 获取当前套接字的属性
        // O_NONBLOCK:非阻塞选项
        fcntl(_sockfd, F_SETFL, flags | O_NONBLOCK);
    }
};


class Poller;
class EventLoop;
class Chennal
{
private:
    int _fd;           // 套接字文件描述符
    uint32_t _events;  // 当前需要监控的事件
    uint32_t _revents; // 当前连接触发的事件
    EventLoop *_loop;  // 所属的Poller
    using EventCallback = std::function<void()>;
    EventCallback _read_callback;  // 可读事件被触发的回调函数
    EventCallback _write_callback; // 可写事件被触发的回调函数
    EventCallback _error_callback; // 错误事件被触发的回调函数
    EventCallback _close_callback; // 连接断开事件被触发的回调函数
    EventCallback _event_callback; // 任意事件被触发的回调函数
public:
    Chennal(EventLoop *loop, int fd) : _fd(fd), _events(0), _revents(0), _loop(loop) {}
    //不能使用默认的析构函数，因为默认的析构函数不会调用基类的析构函数，导致基类的析构函数不会被调用，导致内存泄漏，也就是这些回调函数不会被释放（置为空）
    ~Chennal()
    {
        _read_callback = nullptr;
        _write_callback = nullptr;
        _error_callback = nullptr; 
        _close_callback = nullptr;
        _event_callback = nullptr;
    }
    int getFd() const
    {
        return _fd;
    }
    uint32_t getEvents() const
    {
        return _events;
    }
    void SetREvents(uint32_t revents)
    {
        _revents = revents;
    }
    void SetReadCallback(const EventCallback &cb) { _read_callback = cb; }
    void SetWriteCallback(const EventCallback &cb) { _write_callback = cb; }
    void SetErrorCallback(const EventCallback &cb) { _error_callback = cb; }
    void SetCloseCallback(const EventCallback &cb) { _close_callback = cb; }
    void SetEventCallback(const EventCallback &cb) { _event_callback = cb; }
    // 判断是否可读
    bool ReadAble() { return _events & EPOLLIN; }
    // 判断是否可写
    bool WriteAble() { return _events & EPOLLOUT; }
    // 启动读事件监控
    void EnableRead()
    {
        _events |= EPOLLIN;
        Update();
    }
    // 启动写事件监控
    void EnableWrite()
    {
        _events |= EPOLLOUT;
        Update();
    }
    // 停止读事件监控
    void DisableRead()
    {
        _events &= ~EPOLLIN;
        Update();
    }
    // 停止写事件监控
    void DisableWrite()
    {
        _events &= ~EPOLLOUT;
        Update();
    }
    // 停止所有事件监控
    void DisableAll()
    {
        _events = 0;
        Update();
    }
    // 移除监控
    void Remove();
    // 修改监控
    void Update();
    // 处理事件,一旦连接触发了事件，就调用这个函数
    void HandleEvent()
    {
        if (_revents & EPOLLIN || _revents & EPOLLRDHUP || _revents & EPOLLPRI)
        {
            if (_read_callback)
                _read_callback();
            
        }
        if (_revents & EPOLLOUT)
        {
            if (_write_callback)
                _write_callback();
        }
        if (_revents & EPOLLERR)
        { 
            
            if (_error_callback)
                _error_callback();
        }
        if (_revents & EPOLLHUP)
        { 
            if (_close_callback)
                _close_callback();
        }
        // 不管任何事件，都调用的回调函数
            if (_event_callback)
                _event_callback(); // 放到事件处理完后调用，刷新活跃度
    }
};

#define MAX_EVENTS 1024
class Poller
{
private:
    int _epfd;                                    // epoll句柄
    struct epoll_event _evs[MAX_EVENTS];          // 事件数组
    std::unordered_map<int, Chennal *> _channels; // 监控的事件集合
private:
    // 对epoll的直接操作
    void Update(Chennal *chennal, int po)
    {
        // eopll_ctl(int epfd,int op,int fd,struct epoll_event *event)
        struct epoll_event ev;
        ev.events = chennal->getEvents();
        ev.data.fd = chennal->getFd();
        int ret = epoll_ctl(_epfd, po, chennal->getFd(), &ev); // 修改或添加事件
        if (ret < 0)
        {
            LOG_ERROR("epoll_ctl error");
        }

        return;
    }
    // 判断一个Chennal是否已经在epoll中
    bool HashChannel(Chennal *chennal)
    {
        if (_channels.find(chennal->getFd()) != _channels.end())
        {
            return true;
        }
        return false;
    }

public:
    Poller()
    {
        _epfd = epoll_create(MAX_EVENTS);
        if (_epfd < 0)
        {
            LOG_ERROR("create epoll failed");
            return;
        }
    }
    // 添加或修改监控事件
    void UpdateEvent(Chennal *chennal)
    {
        if (!HashChannel(chennal))
        {
            _channels.insert(std::make_pair(chennal->getFd(), chennal));
            Update(chennal, EPOLL_CTL_ADD);
        }
        else
        {
            Update(chennal, EPOLL_CTL_MOD); // 修改
        }
    }
    // 删除监控事件
    void RemoveEvent(Chennal *chennal)
    {
        auto it = _channels.find(chennal->getFd()); // 找到对应的Chennal
        if (it != _channels.end())
        {
            _channels.erase(it); // 删除
        }
        Update(chennal, EPOLL_CTL_DEL); // 删除
    }
    // 开始监控
    void Poll(std::vector<Chennal *> *active)
    {
        int nfds = epoll_wait(_epfd, _evs, MAX_EVENTS, -1); // 等待事件发生 返回活跃的事件数
        if (nfds < 0)
        {
            if (errno == EINTR)
            {
                return;
            }
            LOG_ERROR("epoll_wait error");
            return;
        }
        for (int i = 0; i < nfds; i++)
        {
            auto it = _channels.find(_evs[i].data.fd); // 找到对应的Chennal
            assert(it != _channels.end());             // 一定存在
            Chennal *chennal = it->second;             // 取出Chennal
            chennal->SetREvents(_evs[i].events);       // 设置当前事件
            active->push_back(chennal);                // 加入活跃集合
        }
    }
};


using TaskFunc = std::function<void()>;
using ReleaseFunc = std::function<void()>;
class TimeTask
{
public:
    TimeTask(uint64_t id, uint32_t timeout, const TaskFunc &cd) : _id(id), _timeout(timeout), _task_cd(cd), _is_running(true)
    {
    }
    ~TimeTask()
    {
        if (_is_running)
            _task_cd();
        _release_cd();
    }
    // 设置定时器释放函数
    void SetReleaseFunc(const ReleaseFunc &cd)
    {
        _release_cd = cd;
    }
    void Cancel() { _is_running = false; }    // 取消定时器任务
    uint32_t DelayTime() { return _timeout; } // 获取定时器延迟时间
private:
    uint64_t _id;            // 定时器任务对象id
    uint32_t _timeout;       // 定时器超时时间
    bool _is_running;        // 定时器任务是否被取消
    TaskFunc _task_cd;       // 定时器任务函数
    ReleaseFunc _release_cd; // 用于删除timeWheel中的定时器对象
};

class TimeWheel
{
public:
    TimeWheel(EventLoop *loop) : _capacity(1024), _tick(0), _wheel(_capacity), _loop(loop), _timerfd(CreateEventFd()), _timer_channel(new Chennal(loop, _timerfd))
    {
        _timer_channel->SetReadCallback(std::bind(&TimeWheel::Ontime, this));
        _timer_channel->EnableRead(); // 启动读事件监控
    }
    ~TimeWheel() {}
    void TimerAdd(uint64_t id, uint32_t timeout, const TaskFunc &cd);// 添加定时器任务

    void TimerDel(uint64_t id) // 删除定时器任务
    {
    }
    void TimerRefresh(uint64_t id); // 刷新、延迟定时任务
    

    void TimerCancel(uint64_t id); // 取消定时器任务
    // 这个接口存在线程安全问题，这个接口实际上不能被外界使用者调用，只能在模块内，在对应的eventloop线程内进行
    bool HasTimer(uint64_t id)
    {
        auto it = _task_map.find(id); // 找到定时器任务对象
        if (it == _task_map.end())
        {
            return false;
        }
        return true;
    }

private:
    void RemoveTimer(uint64_t id)
    {
        auto it = _task_map.find(id); // 找到定时器任务对象
        if (it != _task_map.end())
        {
            _task_map.erase(it); // 从任务映射表中删除定时器任务对象
        }
    }
    int CreateEventFd()
    {
        int fd = timerfd_create(CLOCK_MONOTONIC, 0);
        if (fd == -1)
        {
            std::cerr << "Error creating timerfd: " << strerror(errno) << std::endl;
            return -1;
        }
        itimerspec its;
        its.it_value.tv_sec = 1;
        its.it_value.tv_nsec = 0;
        its.it_interval.tv_sec = 1;
        its.it_interval.tv_nsec = 0;
        int ret = timerfd_settime(fd, 0, &its, NULL);
        if (ret == -1)
        {
            std::cerr << "Error setting timerfd: " << strerror(errno) << std::endl;
            return -1;
        }
        return fd;
    }

    int ReadTimefd()
    {
        uint64_t exp;
        ssize_t n = read(_timerfd, &exp, sizeof(exp));
        if (n == -1)
        {
            std::cerr << "Error reading timerfd: " << strerror(errno) << std::endl;
            abort();
        }
        // std::cout << "Timer expired " << exp << " times" << std::endl;
        // LOG_INFO("Timer expired: %ld times", exp);
        return exp;
    }
    // 这个函数应该每秒被执行一次，相当于秒针向后走一步
    void RunTimerTask() // 执行定时器任务
    {
        _tick = (_tick + 1) % _capacity; // 时间轮指针向后走一步
        _wheel[_tick].clear();           // 清空指定位置的数组，就会把数组中保存的所有管理定时器对象的shared_ptr释放掉释放掉
    }

    void Ontime()
    {
        int times=ReadTimefd();
        for(int i=0;i<times;i++){
            RunTimerTask();
        }
    }

    void TimerAddr(uint64_t id, uint32_t timeout, const TaskFunc &cd) // 添加定时器任务
    {
        Ptrtask pt(new TimeTask(id, timeout, cd));                        // 创建定时器对象
        pt->SetReleaseFunc(std::bind(&TimeWheel::RemoveTimer, this, id)); // 设置定时器释放函数
        int pos = (_tick + timeout) % _capacity;                          // 计算定时器位置
        _wheel[pos].push_back(pt);                                        // 保存定时器对象到指定位置的数组中
        _task_map[id] = pt;                                               // 保存定时器对象到任务映射表
    }

    void TimerRefreshr(uint64_t id) // 刷新、延迟定时任务
    {
        auto it = _task_map.find(id); // 找到定时器任务对象
        if (it == _task_map.end())
        {
            return;
        }
        Ptrtask pt = it->second.lock();        // lock()获取共享指针(weak_ptr)管理的对象对应的shared_ptr
        int delay = pt->DelayTime();           // 获取定时器延迟时间
        int pos = (_tick + delay) % _capacity; // 计算定时器位置
        _wheel[pos].push_back(pt);             // 保存定时器对象到指定位置的数组中
    }

    void TimerCancelr(uint64_t id) // 取消定时器任务
    {
        auto it = _task_map.find(id); // 找到定时器任务对象
        if (it == _task_map.end())
        {
            return;
        }
        Ptrtask pt = it->second.lock(); // lock()获取共享指针(weak_ptr)管理的对象对应的shared_ptr
        if (pt)
            pt->Cancel();
    }

private:
    using WeakTask = std::weak_ptr<TimeTask>;         // 弱引用
    using Ptrtask = std::shared_ptr<TimeTask>;        // 共享指针
    int _tick;                                        // 时间轮指针 ,当前指针走到哪里释放哪里,释放哪里，就相当于执行哪里任务
    int _capacity;                                    // 时间轮容量---最大延迟时间
    std::vector<std::vector<Ptrtask>> _wheel;         // 时间轮
    std::unordered_map<uint64_t, WeakTask> _task_map; // 任务id与任务对象映射表

    EventLoop *_loop;                        // 所属的EventLoop
    int _timerfd;                            // 定时器fd
    std::unique_ptr<Chennal> _timer_channel; // 定时器fd对应的Chennal
};



class EventLoop
{
private:
    using Functor = std::function<void()>;
    std::thread::id _thread_id;              // 线程id
    int _event_fd;                           // eventfd唤醒IO事件监控有可能导致的阻塞
    std::unique_ptr<Chennal> _event_channel; // eventfd对应的Chennal
    Poller _poller;                          // 进行所有描述符的事件监控
    std::vector<Functor> _tasks;             // 任务池
    std::mutex _mutex;                       // 互斥锁
    TimeWheel _timer_wheel;                  // 定时器管理器
public:
    // 处理任务池中的所有任务
    void RunAllTasks()
    {
        std::vector<Functor> tmp_tasks;
        {
            std::lock_guard<std::mutex> lock(_mutex);
            tmp_tasks.swap(_tasks);
        }
        for (auto &task : tmp_tasks)
        {
            task();
        }
    }
    static int CreateEventFd()
    {
        int event_fd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
        if (event_fd < 0)
        {
            LOG_ERROR("create eventfd failed");
            return -1;
        }
        return event_fd;
    }
    void ReadEventfd()
    {
        uint64_t res = 0;
        int ret = read(_event_fd, &res, sizeof(res));
        if (ret < 0)
        {
            if (errno == EAGAIN || errno == EINTR)
            {
                return;
            }
            LOG_ERROR("read eventfd failed");
            return;

            /* code */
        }
    }
    void WeakUpEventfd()
    {
        uint64_t val = 0;
        int ret = write(_event_fd, &val, sizeof(val));
        if (ret < 0)
        {
            if (errno == EAGAIN || errno == EINTR)
            {
                return;
            }
            LOG_ERROR("write eventfd failed");
            return;
        }
        return;
    }

public:
    EventLoop() : _thread_id(std::this_thread::get_id()),
                  _event_fd(CreateEventFd()),
                  _event_channel(new Chennal(this, _event_fd)),
                  _timer_wheel(this)
    {
        // 给eventfd添加可读事件回调函数，读取eventfd事件通知次数
        _event_channel->SetReadCallback(std::bind(&EventLoop::ReadEventfd, this));
        // 启动eventfd的读事件监控
        _event_channel->EnableRead();
    }
    // 用于判断当前线程是否是EventLoop对应的线程
    bool IsInLoop()
    {
        return std::this_thread::get_id() == _thread_id;
    }
    void AssertInLoop()
    {
        assert(std::this_thread::get_id() == _thread_id);
    }
    // 判断将要执行的任务是否处于当前线程中，如果是则执行，不是则压入队列
    void RunInLoop(const Functor &cb)
    {
        if (IsInLoop())
        {
            return cb();
        }
        return QueueInLoop(cb);
    }
    // 将操作放入队列(任务池)，等待下次IO事件唤醒
    void QueueInLoop(const Functor &cb)
    {
        {
            std::lock_guard<std::mutex> lock(_mutex);
            _tasks.push_back(cb);
        }
        // 唤醒有可能因为没有事件就绪，而导致的epoll阻塞
        // 其实就是给eventfd写入一个数据，eventfd就会触发可读事件
        WeakUpEventfd();
    }

    // 添加/修改描述符的事件监控
    void UpdateEvent(Chennal *chennal) { _poller.UpdateEvent(chennal); }
    // 删除描述符的事件监控
    void RemoveEvent(Chennal *chennal) { _poller.RemoveEvent(chennal); }
    void AddTimer(uint64_t id, uint32_t timeout, const TaskFunc &cd) { _timer_wheel.TimerAdd(id, timeout, cd); }
    void RefreshTimer(uint64_t id) { _timer_wheel.TimerRefresh(id); }
    void CancelTimer(uint64_t id) { _timer_wheel.TimerCancel(id); }
    bool HasTimer(uint64_t id) { return _timer_wheel.HasTimer(id); }
    // 三步走：事件监控，任务处理，执行任务
    void Start()
    {
        while(1){
                // 事件监控
            std::vector<Chennal *> active;
            _poller.Poll(&active);
            // 任务处理
            for (auto &chennal : active)
            {
                chennal->HandleEvent();
            }
            // 执行任务
            RunAllTasks();
        }
    }
};


class Any
{
private:
    class holder
    {
    public:
        virtual ~holder() {}
        virtual const std::type_info& type() = 0;
        virtual holder *clone() = 0;
    };

    template <class T>
    class placeholder : public holder
    {
    public:
        placeholder(const T &val) : _val(val) {}
        // ~placeholder() {
        //     // std::cout << "placeholder<T>析构: "<< std::endl;
        // }
        // 获取子类对象保存的数据类型
        virtual const std::type_info& type()
        {
            return typeid(T);
        }
        // 克隆当前的对象自身，返回一个新的对象
        virtual holder *clone()
        {
            return new placeholder<T>(_val);
        }

    public:
        T _val;//保存的数据
    };
    holder *_content;

public:
    Any() : _content(nullptr) {}
    template <class T>
    Any(const T &val) : _content(new placeholder<T>(val)) {}
    Any(const Any &other) : _content(other._content ? other._content->clone() : nullptr) {}
    
    ~Any()
    {
        delete _content;
    }

    Any &swap(Any &other)
    {
        std::swap(_content, other._content);
        return *this;
    }

    template <class T>
    // 返回子类对象保存的数据的指针
    T *get()
    {
        // 想要获取的数据类型，必须和保存的数据类型一致
        assert(typeid(T) == _content->type());
        return &((placeholder<T> *)_content)->_val;
    }
    template <class T>
    // 赋值运算符的重载函数
    Any &operator=(const T &val)
    {
        // 为val构造一个临时的通用容器，然后与当前容器自身进行指针交换，临时对象释放的时候，原来的数据就也被释放了
        Any(val).swap(*this);
        return *this;
    }
    Any &operator=(const Any &other)
    {
        Any(other).swap(*this);
        return *this;
    }
};

//DISCONECTED 连接关闭状态
//CONNECTING 连接建立状态
//CONNECTED 连接成功状态
//DISCONNECTING 连接断开状态 待关闭状态

class Connection;

typedef enum{
    DISCONECTED,CONNECTING,CONNECTED,DISCONNECTING
}ConnStatu;
using PtrConnection=std::shared_ptr<Connection>;
class Connection : public std::enable_shared_from_this<Connection>{
    private:
        uint64_t _conn_id; // 连接的唯一id，便于连接的管理和查找
        int _sockfd;       // 连接关联的文件描述符
        bool _enable_inactive_release; // 连接是否启动非活跃销毁的判断标志，默认为false
        EventLoop* _loop; // 所属的EventLoop
        ConnStatu _statu;  // 连接状态
        Socket _socket;   // 套接字操作管理 
        Chennal _chennal;  // 连接对应的事件监控对象
        Buffer _in_buffer; // 输入缓冲区
        Buffer _out_buffer; // 输出缓冲区
        Any _context;      // 请求的接收处理上下文


        //这个四个回调函数，是让服务器模块来设置的(其实服务器模块的处理回调也是组件使用者设置的)
        //换句话说，这几个回调都是组件使用者使用的
        using ConnectedCallback=std::function<void(const PtrConnection&)>;
        using MessageCallback=std::function<void(const PtrConnection&,Buffer *)>;
        using CloseCallback=std::function<void(const PtrConnection&)>;
        using AnyEventCallback=std::function<void(const PtrConnection&)>;
        ConnectedCallback _connected_callback;
        MessageCallback _message_callback;
        CloseCallback _close_callback;
        AnyEventCallback _any_event_callback;

        //组件内的连接关闭回调--组件内设置的，因为服务器组件内会把所有的连接管理起来
        //一旦某个连接要关闭，就一个从管理的地方移除掉自己的信息
        CloseCallback _server_closed_callback;
    private:
        //五个Channel的事件回调函数
        //描述符可读事件触发后调用的函数，接收socket数据放到接收缓冲区中，然后调用_messag_callback回调函数
        void HandelRead(){
            //1 接收socket的数据，放到缓冲区中
            char data[65536];
            int ret=_socket.RecvNonBlock(data,65536);
            if(ret<0){
                return ShutdownInLoop();
            }
            //这里如果等于0表示的是没有读取到数据，而并不是连接断开了
            //将数据放入输入缓冲区
            _in_buffer.write(data,ret);
            //2 调用_message_callback回调函数，处理接收到的消息
            if(_in_buffer.readable_size()>0)
            {
                //shared_from_this() --从当前对象自身获取自身的shared_ptr管理对象
                return _message_callback(shared_from_this(),&_in_buffer);
            }
        }
        //描述符可写事件触发后调用的函数，从发送缓冲区取出数据，然后发送到socket，然后调用_write_complete_callback回调函数
        void HandelWrite(){
            //_out_buffer中保存的数据就是要发送的数据
            size_t ret=_socket.SendNonBlock(_out_buffer.current_read_pos(),_out_buffer.readable_size());
            if(ret<0){
                //发送错误就关闭连接
                if(_in_buffer.readable_size()>0){
                    _message_callback(shared_from_this(),&_in_buffer);

                }
                return Release();//这时候就是实际的关闭释放操作了

            }
            _out_buffer.move_read_pos(ret);//千万不要忘记移动读指针
            if(_out_buffer.readable_size()==0){
                _chennal.DisableWrite();//没有数据待发生了，关闭写事件监控
                //如果当前是连接关闭状态，则有数据，发送完整数据释放连接，没有数据则直接释放
                if(_statu==DISCONNECTING){
                    return Release();
                }
            }
            return ;
        }
        //描述符错误事件触发后调用的函数，关闭连接，然后调用_close_callback回调函数
        void HandelError(){
            if(_in_buffer.readable_size()>0){
                    _message_callback(shared_from_this(),&_in_buffer);
            }
            return Release();//这时候就是实际的关闭释放操作了
        }
        //描述符关闭事件触发后调用的函数，清理连接资源，然后调用_server_closed_callback回调函数
        void HandelClose(){
            return HandelError();
        }
        //描述符任意事件触发后调用的函数，调用_any_event_callback回调函数
        void HandelEvent(){
            // 1. 刷新连接的活跃度--延迟定时销毁任务
            if(_enable_inactive_release==true){
                _loop->RefreshTimer(_conn_id);
            }
            // 2. 调用组件使用者的任意事件回调函数
            if(_any_event_callback){_any_event_callback(shared_from_this());}
        }
         //连接建立成功后调用的函数，设置连接状态为CONNECTED，然后调用_connected_callback回调函数
        void EstablishedInLoop(){
            assert(_statu==CONNECTING);
            _statu=CONNECTED;
            _chennal.EnableRead(); 
            if(_connected_callback){_connected_callback(shared_from_this());}
        }
        //这个接口才是实际的释放接口
        void ReleaseInLoop(){
            //1. 修改连接状态为DISCONECTED
            _statu=DISCONECTED;
            //2. 移除连接的事件监控
            _chennal.Remove();
            //3. 关闭描述符
            _socket.Close();
            //4. 如果当前定时器队列还有定时销毁任务，则取消任务
            if(_loop->HasTimer(_conn_id)) CancelInactiveReleaseInLoop();
            //5. 调用关闭回调函数，避免先移除服务器管理的连接信息导致Connection释放，再去处理会出错，因此先调用用户
            if(_close_callback){_close_callback(shared_from_this());}
            //6. 调用服务器关闭回调函数，通知服务器管理器，连接已经关闭
            if(_server_closed_callback){_server_closed_callback(shared_from_this());}
        }
        //这个接口并不是实际的发送接口，而是把数据放到了发送缓冲区，启动了可写事件监控
        void SendInLoop( Buffer &buf){
            if(_statu==DISCONECTED){return;}
            // _out_buffer.write(data,len);
            _out_buffer.writeAsBuffer(buf);
            if(_chennal.WriteAble()==false){
                _chennal.EnableWrite();
            }
        }
        //这个关闭操作并非实际的连接释放操作，需要判断还有没有数据待处理，待发送
        void ShutdownInLoop(){
            _statu=DISCONNECTING;//设置连接为半关闭状态
            if(_in_buffer.readable_size()>0){
                if(_message_callback) _message_callback(shared_from_this(),&_in_buffer);
            }
            //要么就是写入数据的时候出错关闭，要么就是没有待发送数据直接关闭
            if(_out_buffer.readable_size()>0){
                if(_chennal.WriteAble()==false){
                    _chennal.EnableWrite();
                }
            }
            if(_out_buffer.readable_size()==0){
                Release();
            }
        }
        //启动非活跃销毁，并定义多长时间无通信就是非活跃，添加定时任务
        void EnableInactiveReleaseInLoop(int sec){
            // 1. 将判断标志_enable_inactive_release设置为true
            _enable_inactive_release=true;
            // 2. 如果当前定时器销毁任务已经cid存在，则只需要刷新即可
            if(_loop->HasTimer(_conn_id)){
                return _loop->RefreshTimer(_conn_id);
            }
            // 3. 否则，添加定时器销毁任务
            _loop->AddTimer(_conn_id,sec,std::bind(&Connection::Release,this));
        }
        //取消非活跃销毁
        void CancelInactiveReleaseInLoop(){
            _enable_inactive_release=false;
            if(_loop->HasTimer(_conn_id)){
                _loop->CancelTimer(_conn_id);
            }
        }
        //切换协议——升级连接，提供新的上下文，新的连接回调，新的消息回调，新的关闭回调，新的任意事件回调
        void UpgradeInLoop(const Any& context,const ConnectedCallback &conn,
                    const MessageCallback &msg,const CloseCallback &close,
                    const AnyEventCallback &any_event){
                        _context=context;
                        _connected_callback=conn;
                        _message_callback=msg;
                        _close_callback=close;
                        _any_event_callback=any_event;
                    }

    public:
        Connection(EventLoop* loop,uint64_t conn_id,int sockfd):_conn_id(conn_id),_sockfd(sockfd),
            _enable_inactive_release(false),_loop(loop),_statu(CONNECTING),_socket(sockfd),_chennal(loop,sockfd)
        {
            _chennal.SetReadCallback(std::bind(&Connection::HandelRead,this));
            _chennal.SetWriteCallback(std::bind(&Connection::HandelWrite,this));
            _chennal.SetErrorCallback(std::bind(&Connection::HandelError,this));
            _chennal.SetCloseCallback(std::bind(&Connection::HandelClose,this));
            _chennal.SetEventCallback(std::bind(&Connection::HandelEvent,this));
        }
        ~Connection() { LOG_DEBUG("RELEASE CONNECTION:%p", this); }
        //获取管理的文件描述符
        int GetFd() const { return _sockfd; }
        //获取连接的唯一id
        uint64_t GetConnId() const { return _conn_id; }
        //获取连接状态
        ConnStatu GetStatu() const { return _statu; }
       //是否处于CONNECTED状态
        bool Connected() { return (_statu == CONNECTED); }
        //设置上下文--连接建立完成进行调用
        void SetContext(const Any &context) { _context = context; }
        //获取上下文，返回的是指针
        Any *GetContext() { return &_context; }
        void SetConnectedCallback(const ConnectedCallback &cb){ _connected_callback=cb; }
        void SetMessageCallback(const MessageCallback &cb){ _message_callback=cb; }
        void SetCloseCallback(const CloseCallback &cb){ _close_callback=cb; }
        void SetAnyEventCallback(const AnyEventCallback &cb){ _any_event_callback=cb; }
        void SetSrvClosedCallback(const CloseCallback &cb){ _server_closed_callback=cb; }
        void Established(){
            _loop->RunInLoop(std::bind(&Connection::EstablishedInLoop,this));
        }

        //发送数据，将数据放到发送缓冲区，启动写事件监控
        void Send(const char* data,size_t len){
            //外界传入的data，可能是个临时的空间，我们现在只是把发送操作压入了任务池，有可能并没有立即执行
            //因此有可能在执行的时候，data已经被释放了，所以这里需要拷贝一份数据，放到发送缓冲区
            Buffer buf;
            buf.write(data,len);
            
            // _loop->RunInLoop(std::bind(&Connection::SendInLoop,this,data,len));
            _loop->RunInLoop(std::bind(&Connection::SendInLoop,this,buf));

        }
        //提供给组件使用者的关闭接口--并不实际关闭，需要判断有没有数据待处理
        void Shutdown(){
            _loop->RunInLoop(std::bind(&Connection::ShutdownInLoop,this));
        }
        void Release(){
            _loop->QueueInLoop(std::bind(&Connection::ReleaseInLoop,this));
        }
        //启动非活跃销毁，并定义多长时间无通信就是非活跃，添加定时任务
        void EnableInactiveRelease(int sec){
            _loop->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop,this,sec));
        }
        //取消非活跃销毁
        void CancelInactiveRelease(){
            _loop->RunInLoop(std::bind(&Connection::CancelInactiveReleaseInLoop,this));
        }
        //切换协议——升级连接，提供新的上下文，新的连接回调，新的消息回调，新的关闭回调，新的任意事件回调
        void Upgrade(const Any& context,const ConnectedCallback &conn,
                    const MessageCallback &msg,const CloseCallback &close,
                    const AnyEventCallback &any_event){
                        _loop->AssertInLoop();
                        _loop->RunInLoop(std::bind(&Connection::UpgradeInLoop,this,context,conn,msg,close,any_event));
                    }
};



class Acceptor{
    private:
        Socket _socket;//用于创建监听套接字
        EventLoop* _loop;//用于对监听套接字进行事件监控
        Chennal _chennal;//用于对监听套接字进行事件管理

        using AcceptCallback=std::function<void(int)>;
        AcceptCallback _accept_callback; // 接收到新连接的回调函数

        private:
            //监听套接字的读事件回调处理函数--获取新连接，调用_accept_callback回调函数进行新连接处理
            void HandelRead(){
                int newfd=_socket.Accept();
                if(newfd<0){
                    return ;
                }
                if(_accept_callback) _accept_callback(newfd);//调用用户设置的接收新连接的回调函数
            }

            int CreateServers(int port){
               bool ret= _socket.CreateServer(port);
                assert(ret==true);
                return _socket.GetFd();
            }
        public:
            //不能将启动读事件监控，放到构造函数中，必须在设置回调函数后，再去启动
            //否则有可能造成启动监控后，立即有事件，处理的时候，回调函数还没设置，导致新连接得不到处理，且资源泄露
            Acceptor(EventLoop* loop ,int port):_socket(CreateServers(port)),_loop(loop),_chennal(loop,_socket.GetFd()){
                _chennal.SetReadCallback(std::bind(&Acceptor::HandelRead,this));//设置监听套接字的读事件回调处理函数
                
            }
            void SetAcceptCallback(const AcceptCallback &cb){_accept_callback=cb;}
            void Listen(){_chennal.EnableRead();}//启动监听套接字的读事件监控}
};


class LoopThread{
    private:
        std::mutex _mutex; // 互斥锁，用于线程间同步
        std::condition_variable _cond; // 条件变量，用于线程间同步
        EventLoop* _loop;    // 线程关联的EventLoop对象
        std::thread _thread;    //EvnentLoop线程对象
    private:
        //实例化EventLoop对象，并开始运行EventLoop模块的功能
        void ThreadEntry(){
            EventLoop loop;
            {
                std::unique_lock<std::mutex> Lock(_mutex);
                _loop=&loop;
                _cond.notify_one(); // 唤醒线程，让线程进入等待状态，等待EventLoop对象初始化完成
            }
            loop.Start(); // 启动EventLoop模块的功能
        }
    public:
        //创建线程，设定线程入口函数
        LoopThread():_loop(NULL),_thread(std::thread(&LoopThread::ThreadEntry,this)){}
        //返回当前线程关联的EventLoop对象指针
        EventLoop* GetLoop(){
            EventLoop* loop=NULL;
            {
                std::unique_lock<std::mutex> Lock(_mutex);
                _cond.wait(Lock,[&](){return _loop!=NULL;});
                loop=_loop;
            }
            return loop;
        }
};


class LoopThreadPool{
    private:
        int _thread_num; // 线程池中线程数量
        int _nex_loop_idx; // 下一个待分配的EventLoop索引
        EventLoop* _base_loop; // 基础EventLoop对象，用于分配给线程,从属线程数量为0，则所有操作都在base_loop中进行
        std::vector<LoopThread*> _threads; // 线程池中线程LoopThread对象
        std::vector<EventLoop*> _loops; // 线程池中EventLoop对象指针集合
    public:
        LoopThreadPool(EventLoop* base_loop):_thread_num(0),_nex_loop_idx(0),_base_loop(base_loop){}
        void SetThreadNum(int num){_thread_num=num;}//设置线程池中线程数量
        void Create(){
            if(_thread_num>0){
                _threads.resize(_thread_num);
                _loops.resize(_thread_num);

                for(int i=0;i<_thread_num;i++){
                    _threads[i]=new LoopThread();
                    _loops[i]=_threads[i]->GetLoop();
                }
            }
            return;
        }//创建从属线程池
        EventLoop* NextLoop(){
            if(_thread_num==0){
                return _base_loop;
            }
            _nex_loop_idx=(_nex_loop_idx+1)%_thread_num;
            return _loops[_nex_loop_idx];
        }
};

class TcpServer {
    private:
        uint64_t _conn_id; //自增的连接id
        int _timeout; // 超时时间,非活跃连接销毁时间
        bool _enable_inactive_release; // 是否启动非活跃销毁功能
        EventLoop _base_loop;//这是主线程的EventLoop对象,负责监听事件处理
        Acceptor _acceptor;//监听套接字的管理对象
        LoopThreadPool _thread_pool;//从属线程池对象，用于分配给线程处理事件
        std::unordered_map<uint64_t,PtrConnection> _conns; // 保存管理所有连接对于的shared_ptr对象
        using ConnectedCallback=std::function<void(const PtrConnection&)>;
        using MessageCallback=std::function<void(const PtrConnection&,Buffer *)>;
        using CloseCallback=std::function<void(const PtrConnection&)>;
        using AnyEventCallback=std::function<void(const PtrConnection&)>;
        using Functor=std::function<void()>;
        ConnectedCallback _connected_callback;
        MessageCallback _message_callback;
        CloseCallback _close_callback;
        AnyEventCallback _any_event_callback;

    private:
        void NewConnection(int fd)//为新连接构造一个Connection对象，并保存到管理器中
        {
            _conn_id++;
            PtrConnection con(new Connection(_thread_pool.NextLoop(),_conn_id,fd));
            con->SetMessageCallback(_message_callback); 
            con->SetCloseCallback(_close_callback); 
            con->SetConnectedCallback(_connected_callback); 
            con->SetAnyEventCallback(_any_event_callback); 
            if(_enable_inactive_release)con->EnableInactiveRelease(_timeout);
            con->Established();
            _conns.insert(std::make_pair(_conn_id,con));
            LOG_DEBUG("New ------------------------");

        }

        void RemoveConnectionInLoop(uint64_t conn_id)//从管理Connection的_conns中移除一个连接
        {
            auto it =_conns.find(conn_id);
            if(it!=_conns.end()){
                _conns.erase(it);
            }
        }
        void RemoveConnection(uint64_t &conn_id)//从管理Connection的_conns中移除一个连接
        {
            _base_loop.RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop,this,conn_id));
        }
        void RunAfterInLoop(const Functor &task,int delay){
            _conn_id++;
            _base_loop.AddTimer(_conn_id,delay,task);
        }
        
    public:
        TcpServer(int port):_conn_id(0),_timeout(0),_enable_inactive_release(false),_acceptor(&_base_loop,port),_thread_pool(&_base_loop)
        {
            _acceptor.SetAcceptCallback(std::bind(&TcpServer::NewConnection,this,std::placeholders::_1));
            _acceptor.Listen();//将监听套接字挂到base_loop中进行监听
        }
        void SetThreadNum(int num){_thread_pool.SetThreadNum(num);}  
        void SetConnectedCallback(const ConnectedCallback &cb){ _connected_callback=cb; }
        void SetMessageCallback(const MessageCallback &cb){ _message_callback=cb; }
        void SetCloseCallback(const CloseCallback &cb){ _close_callback=cb; } 
        void SetAnyEventCallback(const AnyEventCallback &cb){ _any_event_callback=cb; }
        void EnableInactiveRelease(int sec){
            _timeout=sec;
            _enable_inactive_release=true;
        }
        void RunAfter(const Functor &task,int delay)//用于添加一个定时任务，在delay毫秒后执行task函数
        {
            _base_loop.RunInLoop(std::bind(&TcpServer::RunAfterInLoop,this,task,delay));
        }
        void Start(){
            _thread_pool.Create(); //创建从属线程池
            _base_loop.Start();//启动主线程的EventLoop模块的功能
        }
};

void Chennal::Remove() { _loop->RemoveEvent(this); }
void Chennal::Update() { _loop->UpdateEvent(this); }

void TimeWheel::TimerAdd(uint64_t id, uint32_t timeout, const TaskFunc &cd) // 添加定时器任务
{
    _loop->RunInLoop(std::bind(&TimeWheel::TimerAddr, this, id, timeout, cd));
}
void TimeWheel::TimerRefresh(uint64_t id) // 刷新、延迟定时任务
{
    _loop->RunInLoop(std::bind(&TimeWheel::TimerRefreshr, this, id));
}
void TimeWheel::TimerCancel(uint64_t id) // 取消定时器任务
{
    _loop->RunInLoop(std::bind(&TimeWheel::TimerCancelr, this, id));
}

class NetWork{
    public:
        NetWork(){
            LOG_DEBUG("NetWork init");
            signal(SIGPIPE, SIG_IGN); // 忽略SIGPIPE信号 防止程序崩溃
        }
};

static NetWork nw; // 全局唯一的NetWork对象
#endif